package com.wudizaba.mqtt.demos.web.detectophone;

import com.alibaba.fastjson.JSONObject;
import com.wudizaba.nb.domain.mqttContext;
import lombok.SneakyThrows;
import org.eclipse.paho.client.mqttv3.*;
import org.eclipse.paho.client.mqttv3.persist.MemoryPersistence;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.stereotype.Component;

import java.io.File;
import java.io.FileOutputStream;
import java.io.OutputStream;
import java.util.Arrays;

@Configuration
public class consumerDetectophone {

    @SneakyThrows
    @Bean
    public void consumer(){
        System.out.println("======模拟MES系统订阅mqtt主题======");
        //创建mqtt客户端
        //MqttClient: 同步调用客户端，使用阻塞方法通信。
        MqttClient client = new MqttClient(mqttContext.BROKER, mqttContext.SUBSCRIBE_CLIENT_ID, new MemoryPersistence());

        //MqttClientPersistence: 代表一个持久的数据存储，用于在传输过程中存储出站和入站的信息，使其能够传递到指定的 QoS。

        //设置连接参数
        //MqttConnectOptions: 连接选项，用于指定连接的参数，下面列举一些常见的方法。
        //setUserName: 设置用户名
        //setPassword: 设置密码
        //setCleanSession: 设置是否清除会话
        //setKeepAliveInterval: 设置心跳间隔
        //setConnectionTimeout: 设置连接超时时间
        //setAutomaticReconnect: 设置是否自动重连
        MqttConnectOptions options = new MqttConnectOptions();
        options.setUserName(mqttContext.USERNAME);
        options.setPassword(mqttContext.PASSWORD.toCharArray());
        options.setConnectionTimeout(60);
        options.setKeepAliveInterval(60);

        //设置回调函数
        client.setCallback(new MqttCallback() {
            //连接丢失时被调用
            @SneakyThrows
            @Override
            public void connectionLost(Throwable throwable) {
                //重连
                System.out.println("======================连接丢失=========================");
                System.out.println(throwable.getMessage());
                client.connect();
                System.out.println("======================重新连接=========================");
                //重连失败就取消重连，直接报错
                if (client.isConnected()) {
                    System.out.println("======================重新连接成功=========================");
                } else {
                    System.out.println("======================重新连接失败=========================");
                    throw new RuntimeException("重新连接失败");
                }
            }

            //收到消息时被调用
            @Override
            public void messageArrived(String s, MqttMessage mqttMessage) throws Exception {
                System.out.println("======================接收消息=========================");
                System.out.println("收到消息：" + Arrays.toString(mqttMessage.getPayload()));
                System.out.println("收到消息的主题：" + s);
                System.out.println("收到消息的Qos：" + mqttMessage.getQos());
                System.out.println("======================接收消息成功=========================");
                //解析消息
                Object parse = JSONObject.parse(new String(mqttMessage.getPayload()));
                if (parse == null) {
                    throw new RuntimeException("消息解析失败");
                }
                System.out.println(parse);
            }

            //消息发送完成时被调用
            @Override
            public void deliveryComplete(IMqttDeliveryToken iMqttDeliveryToken) {
                System.out.println("======================消息发送=========================");
                System.out.println(iMqttDeliveryToken.isComplete());
                System.out.println("======================消息发送成功=========================");

            }
        });

        client.connect(options);

        //订阅
        client.subscribe(mqttContext.TOPIC, mqttContext.QoS_2);
        System.out.println("======================订阅成功=========================");
    }

}
